
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：数据缓存服务器
----------------------------------------------------------------
--- 1 从指定的 ssdb 中读取数据 scan模式
--- 2 写入到指定mysql dc中去
-------------------------------------------------------------------



local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

local string_char  = string.char
local string_sub   = string.sub
local string_byte  = string.byte
local string_gsub  = string.gsub
local string_find  = string.find
local string_upper = string.upper

local math_fmod   = math.fmod
local math_floor  = math.floor
local math_pow    = math.pow
local math_abs    = math.abs


local uu           = require( "modules.commhm.utils" )
local log          = require( "modules.commhm.log" )
local ns_time      = require( "modules.commhm.time" )
local ns_network   = require( "modules.commhm.network" )
local ns_http      = require( "modules.commhm.resty.http" )

local gg               = require( "modules.data_cache.scan_sync_gg" )
local ns_config        = require( "modules.data_cache.scan_sync_config")

local CNgxTimerWrap    = require("modules.commhm.ngxTimerWrap" )
local ns_network       = require("modules.commhm.network");
local ns_env           = require("modules.env.env");

local ns_CProgressMgr  = require("modules.commhm.framework.CProgressMgr");

--local ns_luaConfigMgr      = require("modules.center_server.luaConfigProxy")
local ns_centerServerProxy   = require("modules.commhm.centerServerProxy");


local FOO = {
	_VERSION = '0.12.1',
	cmds = {},

	worker_id = ngx.worker.id(),

	lock_last_time = 0,
	stat = 0,

	uin_step = 100,

	-- scan_progress = 扫描文件管理
	-- info { select=0, now="", }

	tj = {},

	mode = 2,     -- 1=read compare only   2=save different

};

if  ns_config.uin_step then
	FOO.uin_step = ns_config.uin_step
end


------------初始化
function FOO.init()
end


-- main 入口
function FOO.init_worker()
	ngx.req.get_headers = function()
		--- protect mode
		return {}
	end

	log.buff_io_wait = -1
	log.day_list_simple( "scan_sync", "init_worker", FOO.worker_id )


	-- scan_sync_map_0.txt
	local file_name_ = "./logs/scan_sync_" .. ns_config.doing_name .. '_' .. ngx.worker.id() .. ".txt"
	FOO.scan_progress = ns_CProgressMgr:new( file_name_ )

	--{{{  配置文件是否都ok
	gg.init();
	if  gg.sync_ssdb_dc_client then
		log.day_list_simple( "scan_sync", "init_worker", FOO.worker_id, "connect ssdb-dc and mysql-dc ok" )
	else
		log.day_list_simple( "scan_sync", "init_worker", FOO.worker_id, "connect ssdb-dc and mysql-dc fail ~~~~~ " )
		return
	end
	gg.final();
	--}}}

	FOO.start_scan();
end



-- 进行一次扫描数据 ( one_timer_tick )
function FOO.scan_tick()
	if  FOO.stat == 2 then      --0=init  1=running  2=ended
		return
	end

	-- 忙闲判断
	-- local load_index_ = 0
	local load_index_ = gg.getMaGateLoad( 'map_scan_sync')
	if  load_index_ > 100 then
		--负载超过100
		log.day_list_simple( "scan_sync", FOO.worker_id, "scan_load_busy", load_index_, uu.to_str( FOO.info ) )
		return
	end

	local now_ = uu.now()
	if  now_ - FOO.lock_last_time < 120 then
		log.day_list_simple( "scan_sync", "scan_tick_busy", FOO.worker_id, load_index_, uu.to_str( FOO.info ) )
		return     --busy
	else
		FOO.lock_last_time = now_
	end

	--打印统计 每10秒
	if  now_ % 10 == 0 then
		local  rp_name_ = "scan_tick_10"
		if  now_ % 300 == 0 then
			-- 每100秒上报一次进度
			rp_name_ = "scan_tick_100"
			FOO.report_progress()
		end
		log.day_list_simple( "scan_sync", rp_name_, FOO.worker_id, load_index_,
				uu.to_str( FOO.info ), uu.to_str(FOO.tj) )
		FOO.tj = {}
	end

	local busy_rate_ =  1 - ( load_index_ / 100 )   ---忙闲比例 ( 1.0 到 0.01 )
	if  busy_rate_ >= 1 then busy_rate_ = 1 end

	local step_limit_ = math.floor(FOO.uin_step * busy_rate_)
	if  step_limit_ < 5 then
		step_limit_ = 5       ---最少扫描5个
	end

	--{{{ 开始一次扫描
	gg.init()
	FOO.scan_one_step( step_limit_ )
	gg.final()
	--}}}

	FOO.save_info_file()
end



----批量处理N个  （hlist扫描1000个）
function FOO.scan_one_step( step_limit_ )

	local select_  = FOO.info.select
	local connect_ = gg.sync_ssdb_dc_client

	if  connect_ then
		local ret_str_ = connect_:hlist( select_, FOO.info.now, "", step_limit_ );
		local ret = table.loadstring( ret_str_ ) or {}
		log.day_list_simple( "scan_sync", "hlist", FOO.worker_id, #ret, step_limit_, FOO.info.now )

		if  #ret == 0 or ( #ret==1 and ret[1]=="ok" ) then
			-- 扫描完成
			log.day_list_simple( "scan_sync", "scan_over", FOO.worker_id, FOO.info.now, select_, #ret )
			FOO.set_all_scan_end( "scan_over" )
		else
			--开始扫描所有key
			for i=1, #ret do
				if  #ret[ i ] < 128 then
					FOO.info.now = ret[ i ]   --记录最后key( key不会太长 )
				else
					log.day_list_simple( "scan_sync", "error_key", FOO.worker_id, FOO.info.now,
							select_, uu.to_str(ret[ i ]) )
				end
				FOO.scan_one_key( connect_, select_, ret[i] )
			end
		end
	end

	FOO.lock_last_time = 0
end



--- 扫描单个key
function FOO.scan_one_key( connect_, select_, key_ )
	--log.day_list_simple( "scan_sync", "scan_one_key", FOO.worker_id, key_ )

	local allret_str_ = connect_:hgetall( select_, key_ );
	local allret_     = table.loadstring( allret_str_ ) or {}

	--计算select 1个ssdb库写到10个mysql库的情况
	local save_select_ = select_
	if  gg.target_select_by_key_mod then
		local key_num_ = tonumber( key_ )
		if  key_num_ then
			save_select_ = key_num_ % gg.target_select_by_key_mod
		end
	end

	if  #allret_ > 200 then
		log.day_list_simple( "scan_sync", "error_scan_v_too_many", FOO.worker_id, key_, #allret_ )
	else
		--log.day_list_simple( "scan_sync", "kv", FOO.worker_id, key_, uu.to_str(allret_) )

		FOO.tj.key = (FOO.tj.key or 0) + 1

		for m=1, #allret_, 2 do
			FOO.tj.hash = (FOO.tj.hash or 0) + 1

			local hash_ = allret_[m]
			if  hash_ and allret_[m+1] and #allret_[m+1]>0 then
				if  gg.sync_mysql_dc_client then

					local read_ret_ = gg.sync_mysql_dc_client:read( save_select_, key_, hash_, nil, nil, 1 )
					if     read_ret_ == allret_[m+1] then
						--ok
						--log.day_list_simple( "scan_sync", "compare_ok", FOO.worker_id, key_, hash_, #read_ret_ )
						FOO.tj.match = (FOO.tj.match or 0) + 1
					else
						-- fail
						if  read_ret_ == ngx.null then
							log.day_list_simple( "scan_sync", "error_compare_null", FOO.worker_id, key_, hash_,
									'\n', allret_[m+1], '\n', 'ngx.null' )
							FOO.tj.null = (FOO.tj.null or 0) + 1
						elseif read_ret_ == '' then
							log.day_list_simple( "scan_sync", "error_compare_len0", FOO.worker_id, key_, hash_,
									'\n', allret_[m+1], '\n', read_ret_ )
							FOO.tj.len0 = (FOO.tj.len0 or 0) + 1
						else
							log.day_list_simple( "scan_sync", "error_compare_fail", FOO.worker_id, key_, hash_,
									'\n', allret_[m+1], '\n', read_ret_ )
							FOO.tj.cfail = (FOO.tj.cfail or 0) + 1
						end

						if  FOO.mode == 2 then
							local save_ret_ = gg.sync_mysql_dc_client:save( save_select_, key_, hash_,
									allret_[m+1], nil, nil, 1 )
							if  save_ret_ == 'OK' then
								FOO.tj.save = (FOO.tj.save or 0) + 1
								log.day_list_simple( "scan_sync", "save_ok", FOO.worker_id, save_select_, key_, hash_ )
							else
								log.day_list_simple( "scan_sync", "error_save_kv", FOO.worker_id,
										save_select_, key_,	allret_[m], #allret_[m+1], save_ret_ )
							end
						end

					end

				else
					log.day_list_simple( "scan_sync", "save_kv_ignore", FOO.worker_id,
							select_, key_, allret_[m], #allret_[m+1] )
				end
			else
				log.day_list_simple( "scan_sync", "error_kv", FOO.worker_id,
						select_, key_, allret_[m], uu.to_str(allret_[m+1]) )
			end
		end
	end

end




----开始扫描----
function FOO.start_scan()
	if  FOO.start_scan_begin then
		log.day_list_simple( "scan_sync", "error_call_start_scan_twice", FOO.worker_id )    --互斥调用
		return
	end
	FOO.start_scan_begin = true

	if  not ( FOO.load_info_file() == 0 ) then
		return
	end

	if  not (FOO.info.select == FOO.worker_id) then
		log.day_list( "scan_sync", FOO.worker_id, "select_not_match_worker_id", FOO.worker_id, FOO.info.select )
		return
	end

	log.day_list_simple( "scan_sync", "start_scan_begin", FOO.worker_id, uu.to_str( FOO.info ) )

	local timer_ = CNgxTimerWrap:new()
	timer_:ngx_timer_every( 1, FOO.scan_tick )
end


----所有任务结尾
function FOO.set_all_scan_end( msg_ )
	FOO.stat = 2   --- 0=init  1=running  2=ended
	log.day_list_simple( "scan_sync", "all_done",  FOO.worker_id, msg_, uu.to_str( FOO.info ) )
end



----加载存盘
function FOO.load_info_file()

	FOO.info = FOO.scan_progress:load()

	if  FOO.info and FOO.info.select and FOO.info.now then
		if  FOO.info.select == "new" then
			FOO.info.select = FOO.worker_id    --全新刷数据 select=new 转换成当前select
		end
		return 0   	--ok
	else
		log.day_list_simple( "scan_sync", "load_info_file_warning", FOO.worker_id, uu.to_str( FOO.info ) )
	end
	return 1
end


----记录存盘
function FOO.save_info_file()
	if  FOO.info then
		FOO.scan_progress:save( FOO.info )
	end
end


-- 上报一次数据
--{["select"]=7,["now"]="469088977"}
function FOO.report_progress( info_ )
	--level_, key_, tt_, value_
	if  FOO.info and FOO.info.select then
		local key_   = 'scan_sync_' .. FOO.info.select
		local value_ = 'select=' .. FOO.info.select .. ',now=' .. FOO.info.now
		ns_centerServerProxy.WWW_fatalReport( 3, key_, 290, value_, 1 )
	end
end


FOO.init()
return FOO;
